package com.ahaoyou.rocketmq;

import com.ahaoyou.compnent.TableHandlerDispatcher;
import com.ahaoyou.compnent.TableHandlerType;
import com.ahaoyou.utils.TableOperate;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Objects;

@Component
@RocketMQMessageListener(topic = "example_canal_topic", consumerGroup = "GroupProducer")
public class CanalConsumer implements RocketMQListener<String> {

    @Autowired
    private TableHandlerDispatcher dispatcher;

    @Override
    public void onMessage(String s) {
        JSONObject jsonObject = JSONObject.parseObject(s);
        //获取到执行的表名
        String table = jsonObject.getString("table");
        //获取到执行的sql语句类型
        String type = jsonObject.getString("type");
        System.out.println(jsonObject.toJSONString());
        if (StringUtils.isBlank(table)
                || StringUtils.isBlank(type)
                || Objects.equals(type, "QUERY")) {
            return;
        }
        TableOperate operate = null;
        switch (type) {
            case "INSERT":
                operate = TableOperate.INSERT;
                break;
            case "UPDATE":
                operate = TableOperate.UPDATE;
                break;
            case "DELETE":
                operate = TableOperate.DELETE;
                break;
            default:
                throw new RuntimeException("表操作类型未知:" + type);
        }
        //执行对应的方法
        dispatcher.dispatch(new TableHandlerType(table, operate), jsonObject);
    }
}